/**
 *
 */
package com.cmcc.pcap.packet.sctp;

import com.cmcc.pcap.packet.capture.JCaptures;
import com.cmcc.pcap.packet.tcp.TcpConnectionCache;
import com.cmcc.pcap.packet.tcp.TcpProducer;
import com.cmcc.pcap.packet.tcp.TcpProtocolDispatcher;
import com.cmcc.pcap.packet.util.CommonStatistics;
import com.cmcc.pcap.packet.util.PacketAnalysisConfig;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.pcap4j.packet.IpPacket;
import org.pcap4j.packet.SctpPacket;
import org.pcap4j.packet.TcpPacket;

import java.util.ArrayList;
import java.util.Map.Entry;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Date 2018年1月11日 下午6:43:37
 * @author Amish
 *
 */
public class SctpDispatcher {
	public static Logger logger = LogManager.getLogger(SctpDispatcher.class);
	JCaptures jCaptures;

	int sctpConnectionCache = 7;//TcpConnectionCache数量，add(生产者线程数量)和get的工作线程的数量(消费者线程的数量)默认与该值相等.TODO 当为0时考虑采用动态弹性的线程数量管理
	int sctpConnectionCacheGetratio = 1;//每个TcpConnectionCache对应的get工作线程数量(消费者线程的数量),即TcpProtocolDispatcher的数量，默认为1，与TcpConnectionCache数量相等
	ConcurrentHashMap<String,DispatcherContext> dispatcherContexts;

	boolean statistic = true;//statistic 为true：启动StatisticTask扫描线程处理的实时统计信息   为false：则不启动.StatisticTask如果启动，需要显示调用TcpDispatcher的stop，来结束线程，否则主线程不会退出TODO 不为true主线程扫描完文件就退出了,如何完善
	boolean statisticStart = false;
	AtomicInteger sctpPacketNum;
	long startTimeLong = 0;//记录所有包处理开始时间
	long stopTimeLong = 0;//记录所有包处理结束时间
	private ScheduledExecutorService  scheduledExecutorService;
	public boolean isOveroad = false;
	int overloadPacketsNum = 0;
	String taskId;

	public SctpDispatcher(JCaptures jCaptures, String taskId){
		this.jCaptures = jCaptures;
        //TODO 读配置文件，初始化各种参数值
		this.sctpPacketNum = new AtomicInteger(0);
		this.dispatcherContexts = new ConcurrentHashMap<String,DispatcherContext>();
		this.sctpConnectionCache = PacketAnalysisConfig.getConfigAsInt("sctp.connection.cache");
		this.sctpConnectionCacheGetratio = PacketAnalysisConfig.getConfigAsInt("sctp.connection.cache.getratio");
		this.overloadPacketsNum = PacketAnalysisConfig.getConfigAsInt("sctp.deal.overload");
		this.taskId = taskId;
	}
	
	private synchronized void startStastics() {
		this.startTimeLong = System.currentTimeMillis();//记录所有包处理开始时间
		this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
		scheduledExecutorService.scheduleAtFixedRate(new StatisticTask(this), 2, 10,TimeUnit.SECONDS);//启动一个定时统计任务
		this.statisticStart = true;
	}


	
	public void dispatcherSctpPacket(IpPacket ipPacket) {
		this.sctpPacketNum.incrementAndGet();
		while(this.isOveroad) {// 过负荷，阻塞
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				logger.error(e.getCause(),e);
			}
		}
		
		if(statistic && !statisticStart){//如果scheduledExecutorService已经启动，将不重复启动
			this.startStastics();
		}
		
		//logger.info(tcpPacket.toString());//收到每个sctp packet 都会打一个消息
		String dispatcherKeyStr = this.gengerteDispatcherKeyStr(ipPacket, sctpConnectionCache);
		logger.debug("dispatcher keyStr is: "+ dispatcherKeyStr );
		
		DispatcherContext dispatcherContext;
		if(dispatcherContexts.containsKey(dispatcherKeyStr)){
			dispatcherContext = dispatcherContexts.get(dispatcherKeyStr);
		}else{
			dispatcherContext = new DispatcherContext(this.sctpConnectionCacheGetratio);//构造方法里会new一个SctpProtocolDispatcher
			dispatcherContext.startSctpProtocolDispatcher();//生成SCTP Packet对应的消费者
			dispatcherContexts.put(dispatcherKeyStr, dispatcherContext);
		}
		
		dispatcherContext.execute(ipPacket);
	}
	
	/**
	 * 通过四元组生成确定的值    TODO IPv4 IPv6如何区分   TODO 找一个更风骚的算法？
	 * @param
	 * @return
	 */
	private String gengerteDispatcherKeyStr(IpPacket ipPacket,int cacheSize){
		
		int src_ip = ipPacket.getHeader().getSrcAddr().hashCode();
		int src_port = ipPacket.get(SctpPacket.class).getHeader().getSrcPort().value();
		int dst_ip = ipPacket.getHeader().getDstAddr().hashCode();
		int dst_port = ipPacket.get(SctpPacket.class).getHeader().getDstPort().value();
		long unionCode = src_ip+src_port+dst_ip+dst_port;
		int dispatcherKeyCode = (int) (Math.abs(unionCode) % cacheSize);
		return Integer.toString(dispatcherKeyCode);
	}
	
	public class DispatcherContext {

		SctpConnectionCache sctpConnectionCache;
		ExecutorService executorService;
		int cacheGetratio = 1;
		//ArrayList<TcpProtocolDispatcher> workers;
		ArrayList<Thread> workerThreads;
		
		public DispatcherContext(int cacheGetratio){
			this.cacheGetratio = cacheGetratio;
			this.sctpConnectionCache = new SctpConnectionCache();
			//this.executorService = Executors.newSingleThreadExecutor();
			this.executorService = new ThreadPoolExecutor(1, 1,
					0L, TimeUnit.MILLISECONDS,
					new LinkedBlockingQueue<Runnable>());//为了获得事件队列的长度，重写了这部分实现.此处为一个单线程线程池，所有提交的任务执行前缓存在无界队列里，可保证提交的任务串行执行。

			//workers = new ArrayList<TcpProtocolDispatcher>();
			workerThreads = new ArrayList<Thread>();
			for(int i = 0 ; i < this.cacheGetratio ; i++){
				SctpProtocolDispatcher worker = new SctpProtocolDispatcher(jCaptures,this.sctpConnectionCache);
				worker.setTaskId(taskId);
				Thread thread = new Thread(worker);
				thread.setDaemon(true);//设置为daemon，在主线程退出时该线程也退出。
				workerThreads.add(thread);
                //workers.add(worker);
			}

		}
		
		public void execute(IpPacket ipPacket) {
			this.executorService.execute(new SctpProducer(ipPacket,this.sctpConnectionCache,jCaptures.timestamp));
		}

		public void startSctpProtocolDispatcher(){//每个DispatcherContext对应一个，每个线程对应一个HTTPXDRGenerateHandler
			for( Thread thread : workerThreads){
				thread.start();
			}
		}
		
		public SctpConnectionCache getSctpConnectionCache(){
			return this.sctpConnectionCache;
		}
		public ExecutorService getExecutorService() {
			return this.executorService;
		}
		
		public void stop() {
			this.executorService.shutdown();
//			tpd.terminate();//此处用Thread.stop可能导致线程无法结束，TOFIX
//			tpd1.terminate();
		}
		
	}

	
	/**
	 * 收集各种统计信息，并输出
	 * @Date 2018年4月19日 下午7:15:41
	 * @author Amish
	 */
	class StatisticTask implements Runnable{
		SctpDispatcher sctpDispatcher;
		public StatisticTask(SctpDispatcher tcpDispatcher) {
			this.sctpDispatcher = sctpDispatcher;
		}
		
		@Override
		public void run() {
			try{
				jCaptures.sctp_packet_exist = true;
				StringBuffer sb = new StringBuffer();
				int queueSizeSum =0;
				int atSum=0;
				int afSum=0;
				int aSum = 0;
				int bSum = 0;
				int cSum = 0;
				int dSum = 0;
				int eSum = 0;
				int fSum = 0;
				sb.append("\n##############################################################################\n");
				sb.append("All Statistic Information [taskid : "+taskId+"]\n");
				sb.append("MainThread receive tcp Packet : "+sctpPacketNum.get()+"\n");
				for(DispatcherContext dispatcherContext : dispatcherContexts.values()){
					ExecutorService es = dispatcherContext.getExecutorService();
					ThreadPoolExecutor tpe = (ThreadPoolExecutor) es;
					int queueSize = tpe.getQueue().size();    //线程池任务队列中未被处理的TCP Packet数量
					queueSizeSum = queueSizeSum + queueSize;
	
					
					SctpConnectionCache sctpConnectionCache = dispatcherContext.getSctpConnectionCache();
					
					int at = sctpConnectionCache.totalConnectionNum.get();          //进入到Cache的SCTP链接数量，等同于收到三次握手的SYN包的数量，重传的SYN不计算
					int af = sctpConnectionCache.finishConnectionNum.get();         //从Cache中取出的SCTP链接数量，这些连接可能是超时、可能是非法协议、可能是被进一步处理等
					int a = sctpConnectionCache.totalSctpPacketNum.get();            //进入到Cache的SCTP Packet数量
					int b = sctpConnectionCache.receiveInvalidSctpPacketNum.get();   //进入到Cache的非法SCTP Packet数量
					int c = sctpConnectionCache.overTimeInvalidSctpPacketNum.get();  //进入到Cache的超时SCTP Packet数量
					int d = sctpConnectionCache.pickupPacketNum.get();              //从Cache中取出的SCTP Packet数量
					int e = a - b - c - d;                                         //Cache中剩余的SCTP Packet数量
					int f = sctpConnectionCache.handledPacketNum.get();             //被工作线程/消费者线程 “处理完”的SCTP Packet数量
					//int g = a - f;
					
					atSum=atSum+at;
					afSum=afSum+af;
					aSum=aSum+a;
					bSum=bSum+b;
					cSum=cSum+c;
					dSum=dSum+d;
					eSum=eSum+e;
					fSum=fSum+f;
	//				aSum=aSum+tcpConnectionCache.totalTcpPacketNum.get();
	//				bSum=bSum+tcpConnectionCache.receiveInvalidTcpPacketNum.get();
	//				cSum=cSum+tcpConnectionCache.overTimeInvalidTcpPacketNum.get();
	//				dSum=dSum+tcpConnectionCache.processedPacketNum.get();
	
					sb.append("SctpConnection : < "+(at-af)+"/"+at+" > "
							+ "| QueueSctpPacket : < "+queueSize+" > "
					        + "| CacheSctpPacket : < "+a+" > "
							+ "| ReceiveInvalidSctpPacket : < "+b+" > "
							+ "| OverTimeInvalidSctpPacket : < "+c+" > "
							+ "| PickupPacket : < "+d+" > "
						    + "| CacheSuplusPacket : < "+e+" > "
					        + "| HandledPacketNum : < "+f+" > \n");
				}
				sb.append("SctpConnection Sum : < "+(atSum-afSum)+"/"+atSum+" > "
						+ "|QueueSctpPacket Sum : < "+queueSizeSum+" > "
						+ "|CacheSctpPacket Sum : < "+aSum+" > "
						+ "| ReceiveInvalidSctpPacket Sum : < "+bSum+" > "
						+ "| OverTimeInvalidSctpPacket Sum : < "+cSum+" > "
						+ "| PickupPacket Sum : < "+dSum+" > "
						+ "| CacheSuplusPacket Sum : < "+eSum+" > "
			       	    + "| HandledPacketNum Sum : < "+fSum+" > \n");
				
				if(overloadPacketsNum != 0){
					checkOverLoad(atSum-afSum,aSum);//触发过负荷检查
				}

				sb.append("\n##############################################################################");
				
				/**
				 * cache里剩余的Packet应该为零
				 *   且
				 * 从cache正常提取出的数据包 和 工作线程处理完成的数据包数量应该相等
				 * 才表示一次任务处理完成
				 */
				if( eSum == 0 && (dSum - fSum) == 0 ) {//TODO 此处可以考虑单独起一个线程做 
					
					sb.append(CommonStatistics.getInstance().HttpTransactionNumToString());//打印一些临时统计信息
					sb.append(CommonStatistics.getInstance().HttpEntityNumToString());//打印一些临时统计信息
					//TEMP 20180831
					//sb.append("\n <"+HttpTransactionSplit.al_length+" >");
					
					sb.append("\n##############################################################################");
					stopTimeLong = System.currentTimeMillis();//记录所有包处理结束时间，误差正负10s，10s为统计任务的周期启动时间
					sb.append("\n  [ all packet deal duration  "+(stopTimeLong-startTimeLong)+"  ms ]");
					sb.append("\n##############################################################################");
					logger.info(sb.toString());
					
					
					//tcpDispatcher.stop();//stop 或调用回调函数通知调用方
//					jCaptures.complete = true;
					jCaptures.sctp_complete = true;
					logger.debug("tcp packets done !");
					
					return;
				}
				logger.info(sb.toString());
			}catch (Exception e){
				logger.error(e.getCause(),e);
			}
		}
		
	}
	
	/**
	 * 简单的过负荷控制，需要重新完善，且只有FileFakeSerial模式下需要过负荷控制
	 * @param tcpConnectionCache
	 */
	public void checkOverLoad(int tcpConnectionCache,int cacheReceiveTcpPacket) {
		boolean fileFakeSerial = "FileFakeSerial".equalsIgnoreCase(PacketAnalysisConfig.getConfigAsString("pcap.deal.mode"));
		if(fileFakeSerial){//只有FileFakeSerial模式下需要过负荷控制
			int todoSctpPacket = sctpPacketNum.get() - cacheReceiveTcpPacket;
			int overload = PacketAnalysisConfig.getConfigAsInt("sctp.deal.overload");
			
			int addthreadhold = PacketAnalysisConfig.getConfigAsInt("sctp.connection.cache.addthreshold");
			int sctpConnectionThread = PacketAnalysisConfig.getConfigAsInt("sctp.connection.cache");
			
			if(sctpConnectionCache == (addthreadhold*sctpConnectionThread) && todoSctpPacket > overload){
				this.isOveroad = true;
				logger.info("Overload control start ,with connection cahe : "+ sctpConnectionCache);
			}else if(this.isOveroad){
				this.isOveroad = false;
				logger.info("Overload control stop ,with connection cahe : "+ sctpConnectionCache);
			}
		}
		
	}

	
	public synchronized void stop(){
		logger.info("[ stoping TcpDispatcher ]");
		if(this.scheduledExecutorService != null)
		    this.scheduledExecutorService.shutdownNow();
		this.statisticStart = false;
		for(Entry<String, DispatcherContext> entry : dispatcherContexts.entrySet()) {
			entry.getValue().stop();
		}
		dispatcherContexts.clear();

		sctpPacketNum.set(0);
		CommonStatistics.getInstance().reset();//之前统计数据清零
		logger.info("[ stoped TcpDispatcher ]");
	}
	
	
}
